package io.gitee.hfl.rocketmq.register;

import io.gitee.hfl.rocketmq.annotation.RocketConsumer;
import io.gitee.hfl.rocketmq.annotation.SubscriptionExpression;
import io.gitee.hfl.rocketmq.consumer.RocketListener;
import io.gitee.hfl.rocketmq.exception.RocketCreateConsumerException;
import io.gitee.hfl.rocketmq.props.RocketProperties;
import io.gitee.hfl.rocketmq.resolver.PropertyResolver;
import io.gitee.hfl.rocketmq.util.StrUtil;
import org.apache.rocketmq.client.apis.ClientConfiguration;
import org.apache.rocketmq.client.apis.ClientException;
import org.apache.rocketmq.client.apis.ClientServiceProvider;
import org.apache.rocketmq.client.apis.consumer.FilterExpression;
import org.apache.rocketmq.client.apis.consumer.PushConsumer;
import org.apache.rocketmq.shaded.com.google.common.collect.ImmutableList;
import org.apache.rocketmq.shaded.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.config.AutowireCapableBeanFactory;
import org.springframework.boot.web.context.WebServerInitializedEvent;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationListener;

import javax.annotation.PreDestroy;
import java.io.IOException;
import java.util.*;
import java.util.stream.Collectors;

/**
 * rocket 的消费者自动注册
 * 使用容器工厂扫描所有consumer, 并根据注解配置属性 subscribe到相应的队列
 *
 * @author hefulin
 * @date 2022/09/07 12:33
 */
public class ConsumerAutoRegister implements ApplicationListener<WebServerInitializedEvent> {

    private final RocketProperties rocketProperties;
    private final ApplicationContext applicationContext;
    private final PropertyResolver propertyResolver;
    private final ClientConfiguration clientConfiguration;
    private final ClientServiceProvider clientServiceProvider;

    private static ImmutableList<PushConsumer> REGISTERED_CONSUMERS;

    public ConsumerAutoRegister(RocketProperties rocketProperties, ApplicationContext applicationContext, PropertyResolver propertyResolver, ClientConfiguration clientConfiguration, ClientServiceProvider clientServiceProvider) {
        this.rocketProperties = rocketProperties;
        this.applicationContext = applicationContext;
        this.propertyResolver = propertyResolver;
        this.clientConfiguration = clientConfiguration;
        this.clientServiceProvider = clientServiceProvider;
    }

    protected Logger log = LoggerFactory.getLogger(this.getClass());

    /**
     * 用来注册consumer的
     */
    public void consumerListenerRegister() {
        AutowireCapableBeanFactory autowireCapableBeanFactory = applicationContext.getAutowireCapableBeanFactory();
        String[] beanNamesForAnnotation = applicationContext.getBeanNamesForAnnotation(RocketConsumer.class);
        List<RocketListener<?>> rocketListeners = new ArrayList<>();
        Arrays.stream(beanNamesForAnnotation)
                .map(x -> (RocketListener<?>) autowireCapableBeanFactory.getBean(x))
                .forEach(x -> {
                    RocketConsumer config = x.getClass().getAnnotation(RocketConsumer.class);
                    if (config.enable() && rocketProperties.isEnable()) {
                        RocketConsumer consumerListener = x.getClass().getAnnotation(RocketConsumer.class);
                        int consumerNums = consumerListener.nums() == 0 ? rocketProperties.getConsumer().getNums() : consumerListener.nums();
                        for (int i = 0; i < consumerNums; i++) {
                            rocketListeners.add(x);
                        }
                    }
                });
        // 注册消费者, 并执行订阅
        this.listenerRegister(rocketListeners.toArray(new RocketListener[0]));
    }

    /**
     * 注册consumerListener
     */
    private void listenerRegister(RocketListener<?>... listener) {
        REGISTERED_CONSUMERS = ImmutableList.copyOf(Arrays.stream(listener).map(x -> {
                    RocketConsumer consumerListener = x.getClass().getAnnotation(RocketConsumer.class);
                    // 消费者组
                    String consumerGroup = propertyResolver.resolvePlaceHolders(consumerListener.group());
                    // 消费者线程数
                    int consumeThreadNums = consumerListener.threadNums() == 0 ? rocketProperties.getConsumer().getThreadsNums() : consumerListener.threadNums();
                    // 匹配模式
                    SubscriptionExpression[] subscriptionExpressions = consumerListener.subscriptionExpressions();
                    Map<String, FilterExpression> subscriptionExpressionMap = new HashMap<>(16);
                    for (SubscriptionExpression subscriptionExpression : subscriptionExpressions) {
                        FilterExpression filterExpression = new FilterExpression(propertyResolver.resolvePlaceHolders(subscriptionExpression.expression()), subscriptionExpression.filterExpressionType());
                        subscriptionExpressionMap.put(propertyResolver.resolvePlaceHolders(subscriptionExpression.topic()), filterExpression);
                    }
                    PushConsumer pushConsumer;
                    try {
                        pushConsumer = clientServiceProvider.newPushConsumerBuilder()
                                .setClientConfiguration(clientConfiguration)
                                .setConsumerGroup(consumerGroup)
                                .setConsumptionThreadCount(consumeThreadNums)
                                .setSubscriptionExpressions(subscriptionExpressionMap)
                                .setMessageListener(applicationContext.getBean(StrUtil.toLowerFirst(x.getClass().getSimpleName()), RocketListener.class))
                                .build();
                    } catch (ClientException e) {
                        log.error("create consumer error group:{}, name:{}", consumerGroup, x.getClass().getSimpleName());
                        e.printStackTrace();
                        throw new RocketCreateConsumerException(String.format("group:%s name:%s failed to create", consumerGroup, x.getClass().getSimpleName()));
                    }
                    log.info("create consumer group:{}, name:{}", consumerGroup, x.getClass().getSimpleName());
                    return pushConsumer;
                })
                .filter(Objects::nonNull)
                .collect(Collectors.toList()));
    }

    @Override
    public void onApplicationEvent(WebServerInitializedEvent webServerInitializedEvent) {
        consumerListenerRegister();
    }

    @PreDestroy
    public void destroy() {
        for (PushConsumer registeredConsumer : REGISTERED_CONSUMERS) {
            try {
                registeredConsumer.close();
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
        log.info("consumers destroyed.");
    }

}
